Arrow: Fix direct memory leak in row lineage vectorized readers by Neuw84 · Pull Request #17296 · apache/iceberg

Neuw84 · 2026-07-19T07:29:18Z

RowIdVectorReader and LastUpdatedSeqVectorReader allocated a fresh BigIntVector from the root allocator on every batch, ignored the reuse parameter, kept no reference to the returned vector and had no-op close() methods. Nothing downstream closes the holders' vectors either, so every batch of a vectorized read that projects _row_id or _last_updated_sequence_number leaked one direct-memory vector.

On long-running Spark row-level operations against format-version 3 tables this grows without bound until the executor is killed (the identical workload against a v2 table runs flat, as the lineage readers are never instantiated).

The readers now own their result vector like the base reader owns vec: they allocate it lazily, reuse it across batches while its capacity suffices and release it in close(), which also closes the delegate readers. New tests assert the root allocator returns to its baseline after close and fail against the previous implementation.

AI Disclosure

Model: Claude Fable
Platform/Tool: Kiro
Human Oversight: reviewed

Relates to #17241

RowIdVectorReader and LastUpdatedSeqVectorReader allocated a fresh BigIntVector from the root allocator on every batch, ignored the reuse parameter, kept no reference to the returned vector and had no-op close() methods. Nothing downstream closes the holders' vectors either, so every batch of a vectorized read that projects _row_id or _last_updated_sequence_number leaked one direct-memory vector. On long-running Spark row-level operations against format-version 3 tables this grows without bound until the executor is killed (the identical workload against a v2 table runs flat, as the lineage readers are never instantiated). The readers now own their result vector like the base reader owns vec: they allocate it lazily, reuse it across batches while its capacity suffices and release it in close(), which also closes the delegate readers. New tests assert the root allocator returns to its baseline after close and fail against the previous implementation.

uros-b

Fix looks correct and well tested, thank you @Neuw84!

Neuw84 · 2026-07-24T18:41:21Z

Thanks for your review!

camper42 · 2026-07-27T04:49:36Z

We deployed Iceberg 1.11.0 with commit 2493bb7 from this PR cherry-picked
into our Spark runtime and observed a substantial improvement in a production
Structured Streaming workload.

Environment/workload:

Spark 3.5.8, Scala 2.12, Java 17
Iceberg format-version 3, merge-on-read
102 concurrent streaming queries performing MERGE operations
20 executors, 4 cores each
16 GiB executor heap + 24 GiB memory overhead (40 GiB pod limit)
spark.sql.shuffle.partitions=400
Same application code, tables, and checkpoints before and after the change

Representative observations:

Before the fix:

Peak executor DirectPoolMemory: 15.43 GiB
Peak container RSS: 33.19 GiB
Peak working set: 33.58 GiB
The application became unhealthy after about 7 minutes, with 54 failed jobs,
314 failed tasks, executor losses, and subsequent shuffle fetch failures.

After cherry-picking this PR:

Peak executor DirectPoolMemory: 0.017 GiB (~17 MiB)
Peak container RSS: 18.05 GiB
Peak working set: 19.31 GiB
After 17 minutes, 755 jobs had succeeded with no failed jobs or tasks.
The executors had processed approximately 771 GiB of shuffle reads and
922 GiB of shuffle writes.
JVM heap pressure remained comparable (15.34 GiB before vs 15.86 GiB after),
indicating that the workload was not materially lighter.

This is a greater than 900x reduction in peak DirectPoolMemory. Container
memory usage can still approach the cgroup limit because of reclaimable
inactive file cache, but RSS and working set remain stable around 16–18 GiB,
with no observed OOMKills.

AI Disclosure

Model: GPT-5.6 Sol
Platform/Tool: Pi
Human Oversight: reviewed

camper42 · 2026-07-27T04:53:40Z

thx @Neuw84 ，#17241 and this fix saved me a lot of time.

pvary · 2026-07-28T08:10:43Z

            ids == null ? null : ArrowVectorAccessors.getVectorAccessor(idsHolder);

-        BigIntVector rowIds = allocateBigIntVector(ROW_ID_ARROW_FIELD, numValsToRead);
+        BigIntVector rowIds = resultVector(numValsToRead);


When reuse is set, then we should use the vectors from there.

Good catch that the parameter is ignored.

I followed the base VectorizedArrowReader.read contract, where reuse is only a signal. It never reads reuse.vector() either, it resets its own vec, and both call sites (ColumnarBatchReader, ArrowBatchReader) pass back the holder this same reader returned, so reuse.vector() is the vector we already cache. Keeping the vector in the reader is also what lets close() release it, which is the actual fix: previously nothing owned those vectors.

I've pushed a change that uses reuse as the same signal the base reader uses (null → reallocate, otherwise reuse the cached vector if capacity suffices), so the parameter is no longer ignored.

Happy to instead adopt reuse.vector() literally if you prefer, but then we need guards for constant/dictionary holders and a decision on who closes a vector the reader did not allocate, that ambiguity is what caused the leak.

The lineage readers ignored the reuse argument and always kept their own result vector. Use reuse the same way the parent reader does: a null holder releases the current vector and allocates a new one, a non-null holder keeps it, and it is reallocated only when the batch no longer fits. The vector stays owned by the reader, which is what lets close() release it, so reading the vector out of the caller's holder would leave ownership ambiguous again. New tests cover both signals: passing the previous holder back reuses the same vector without growing allocated memory, and a larger batch grows it while releasing the previous one.

pvary · 2026-07-28T10:38:40Z

+          vec.close();
+        }
+
+        this.vec = allocateBigIntVector(ROW_ID_ARROW_FIELD, numValsToRead);


We could allocate batchSize sized vectors, and then we don't need to check the size every time.

Fixed on the latest push. However, note that we added a new field.

Both readers now track the batch size in setBatchSize (with the same
DEFAULT_BATCH_SIZE fallback the parent uses) and allocate the result vector for a full batch, so
resultVector no longer compares capacities. Growing the batch size releases the undersized vector
so the next read allocates one that fits, which keeps setRowGroupInfo/setBatchSize reuse safe.

pvary · 2026-07-28T10:39:24Z

Don't we have the same problem with LastUpdatedSeqVectorReader?

Both lineage readers sized their result vector for the current batch and compared the capacity on every read. Track the batch size like the parent reader does and allocate for a full batch instead, so reads no longer check the capacity: a shorter final batch fits, and a larger batch size releases the undersized vector so the next read reallocates.

Neuw84 · 2026-07-28T11:17:24Z

Yes, it had exactly the same leak, and it is fixed in this PR. The two readers are changed
symmetrically. LastUpdatedSeqVectorReader also allocated a fresh BigIntVector from the root
allocator on every read (allocateBigIntVector(LAST_UPDATED_SEQ, numValsToRead)), kept no
reference to it and had a no-op close(), so it leaked one vector per batch whenever
_last_updated_sequence_number was projected.

Thanks for your review!.

pvary · 2026-07-28T11:35:20Z

    private final VectorizedReader<VectorHolder> posReader;
    private NullabilityHolder nulls;
+    private BigIntVector vec;
+    private int resultBatchSize = DEFAULT_BATCH_SIZE;


nit: Could we call it batchSize as we do in the other readers?

pvary · 2026-07-28T11:36:28Z

+      if (reuse == null || vec == null) {
+        if (vec != null) {
+          vec.close();
+        }
+
+        this.vec = allocateBigIntVector(ROW_ID_ARROW_FIELD, resultBatchSize);
+      } else {
+        vec.setValueCount(0);
+      }


nit:
This is a bit easier to read

Suggested change

if (reuse == null || vec == null) {

if (vec != null) {

vec.close();

}

this.vec = allocateBigIntVector(ROW_ID_ARROW_FIELD, resultBatchSize);

} else {

vec.setValueCount(0);

}

if (reuse != null && vec != null) {

vec.setValueCount(0);

} else {

if (vec != null) {

vec.close();

}

this.vec = allocateBigIntVector(RowIdVectorReader.ROW_ID_ARROW_FIELD, resultBatchSize);

}

pvary · 2026-07-28T11:38:46Z

+import org.apache.iceberg.arrow.ArrowAllocation;
+import org.junit.jupiter.api.Test;
+
+public class TestLineageVectorReaders {


package private.

Maybe call it TestVectorizedArrowReader - this is where I would look for any test for VectorizedArrowReader

pvary · 2026-07-28T11:39:09Z

+  private static final int NUM_BATCHES = 64;
+
+  @Test
+  public void testRowIdReaderReleasesMemoryOnClose() {


package private, and remove the test from the test method name

pvary · 2026-07-28T11:42:22Z

+        this.vec = null;
+      }
+
+      this.resultBatchSize = (batchSize == 0) ? DEFAULT_BATCH_SIZE : batchSize;


This is very strange to me.
Do you happen to know why we create 0 sized vector, and nulls?
Should we do this the other way around, and create DEFAULT_BATCH_SIZE stuff if the user sets 0?

For context on where the pattern comes from: PositionVectorReader.setBatchSize has the same ordering (holder sized from the argument, batchSize resolved afterwards), and the parent VectorizedArrowReader.setBatchSize resolves the value but sizes NullabilityHolder from the resolved field only later in read. Happy to clean up PositionVectorReader in this PR too if you want it consistent, or leave it for a follow-up since it is unrelated to the leak.

We might want to ask @amogh-jahagirdar to review.
He might have more context as he was the one who added the last readers.

Resolve the batch size before it is used, so a zero batch size no longer sizes the nullability holder and the result vector differently. Name the field batchSize like the other readers in this class, invert the branch in resultVector as suggested, and rename the test to TestVectorizedArrowReader with package private visibility and method names without the test prefix.

Neuw84 · 2026-07-28T12:06:43Z

Addressed all your comments

Neuw84 · 2026-07-28T13:36:10Z

@amogh-jahagirdar, if you have some time please look at the comments.

I discovered a leak (fixed in this PR) but we have some doubts on the previous logic on the Arrow readers for row lineage readers.

Let us know!, my opinion is that if the change ( from your comments) is big, lets merge this as is critical ( basically V3 on Structured Streaming is broken), I will open another one for what is suggested... if it is small change after you clarification I will implement the changes on this one.

PositionVectorReader sized its nullability holder from the batch size argument but resolved a zero batch size to DEFAULT_BATCH_SIZE only afterwards, so a zero batch size paired a default sized vector with an empty holder. NullabilityHolder.isNullAt indexes its array without a bounds check, so every null check on the returned holder then failed while the vector reported a full batch of values. Resolve the batch size first and size the holder from the resolved value, as the row lineage readers now do. New tests read a batch after setting a zero batch size and assert the holder is sized like the vector for both the position reader and the row id reader.

Neuw84 · 2026-07-28T15:48:25Z

Have been tinkering a little bit more on that 0.

I went looking for where a zero batch size can come from, because the answer decides whether the
== 0 guard should exist at all. What I found:

A zero batch size is reachable, and only ever from an explicit caller. There is no validation
anywhere on the path:

Parquet.ReadBuilder.recordsPerBatch(int) - no check, default 10000.
new ArrowReader(TableScan, int batchSize, boolean) - public API, no check, passes straight through
to recordsPerBatch.
Spark: read.parquet.vectorization.batch-size / the matching SQL conf, default
PARQUET_BATCH_SIZE_DEFAULT = 5000, flows through SparkReadConf.parquetBatchSize() ->
ParquetBatchReadConf -> recordsPerBatch. A user setting 0 is accepted.

DEFAULT_BATCH_SIZE is also 5000, the same value as PARQUET_BATCH_SIZE_DEFAULT, so the guard reads
like "treat 0 as unset and fall back to the default".

But a zero batch size cannot actually work end to end. In
VectorizedParquetReader.FileIterator.next():

int numValuesToRead = (int) Math.min(nextRowGroupStart - valuesRead, batchSize);
...
valuesRead += numValuesToRead;

with batchSize == 0 this is always 0, valuesRead never advances and hasNext() stays true, so
the iterator returns empty batches forever. The iterator uses its own batchSize from ReadConf, so
the readers' == 0 guard cannot rescue that; it only prevents zero-capacity allocations inside the
readers.

What the guard does buy is avoiding a much worse failure, and that is where the ordering matters.
NullabilityHolder.isNullAt indexes a raw array with no bounds check:

public byte isNullAt(int index) {
  return isNull[index];
}

and both ColumnVector (arrow) and IcebergArrowColumnVector (Spark) call it through
holder.nullabilityHolder(). With the old ordering, a zero batch size produced a DEFAULT_BATCH_SIZE
sized vector paired with a zero length holder, so the vector reports 5000 values while any null check
on it throws ArrayIndexOutOfBoundsException. Resolving the size first keeps the two consistent.

So my read is: the guard is worth keeping as defence in depth, the inconsistency was the real
problem, and the actual fix for a user passing 0 belongs at the entry points. Two follow-ups I would
suggest, both out of scope of this pull request ( as I see it):

Reject a non-positive batch size in Parquet.ReadBuilder.recordsPerBatch and the ArrowReader
constructor, since it can only produce an infinite stream of empty batches. That would let the
== 0 guards go away entirely.
VectorizedArrowReader.setBatchSize (the parent, line 136) resolves the value into its own field
but passes the raw argument to vectorizedColumnIterator.setBatchSize(batchSize). Same
inconsistency, one level up. I left it alone because changing what the column iterator receives is
a behaviour change unrelated to this leak.

Let me know your opinions, happy apply the changes on those classes in this pull request.

github-actions Bot added the arrow label Jul 19, 2026

Neuw84 mentioned this pull request Jul 19, 2026

v3 deletion-vector MERGE grows executor memory without bound under high-churn streaming upserts #17241

Open

3 tasks

uros-b approved these changes Jul 23, 2026

View reviewed changes

pvary reviewed Jul 28, 2026

View reviewed changes

Neuw84 requested a review from pvary July 28, 2026 08:31

pvary reviewed Jul 28, 2026

View reviewed changes

Merge branch 'apache:main' into fix-lineage-reader-vector-leak

a63496f

Neuw84 requested a review from pvary July 28, 2026 11:20

pvary reviewed Jul 28, 2026

View reviewed changes

Neuw84 requested a review from pvary July 28, 2026 12:06

Uh oh!

Conversation

Neuw84 commented Jul 19, 2026

Uh oh!

uros-b left a comment

Choose a reason for hiding this comment

Uh oh!

Neuw84 commented Jul 24, 2026

Uh oh!

camper42 commented Jul 27, 2026

Uh oh!

camper42 commented Jul 27, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Neuw84 Jul 28, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Neuw84 Jul 28, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

pvary commented Jul 28, 2026

Uh oh!

Neuw84 commented Jul 28, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Neuw84 Jul 28, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Neuw84 commented Jul 28, 2026

Uh oh!

Neuw84 commented Jul 28, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Neuw84 commented Jul 28, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Neuw84 Jul 28, 2026 •

edited

Loading

Neuw84 Jul 28, 2026 •

edited

Loading

Neuw84 commented Jul 28, 2026 •

edited

Loading

Neuw84 Jul 28, 2026 •

edited

Loading

Neuw84 commented Jul 28, 2026 •

edited

Loading

Neuw84 commented Jul 28, 2026 •

edited

Loading